import schedule
import time
import threading
from datetime import datetime
from models.collaborative_filtering.item_cf import ItemCF
from models.content_based.content_based_improved import ContentBasedRecommender
from models.hybrid.hybrid_recommender import HybridRecommender
from utils.db import get_redis_conn, close_all_connections, get_mysql_conn
from mlxtend.frequent_patterns import apriori, association_rules
import pandas as pd
import numpy as np
import json
from config.config import RECOMMENDATION_CONFIG
import mysql.connector
import redis
import logging
import os

class ModelUpdater:
    """模型更新类，负责定时更新各种推荐模型"""
    
    def __init__(self):
        """初始化模型更新器"""
        self.mysql_conn = None
        self.redis_conn = None
        self.item_cf = ItemCF()
        self.content_recommender = ContentBasedRecommender()
        self.hybrid_recommender = HybridRecommender()
        self.running = False
        self.scheduler_thread = None
        
        self._init_connections()
    
    def _init_connections(self):
        """初始化数据库连接"""
        try:
            # 初始化MySQL连接
            self.mysql_conn = mysql.connector.connect(
                host=os.getenv('MYSQL_HOST', 'localhost'),
                user=os.getenv('MYSQL_USER', 'root'),
                password=os.getenv('MYSQL_PASSWORD', ''),
                database=os.getenv('MYSQL_DATABASE', 'collector')
            )
            
            # 初始化Redis连接
            self.redis_conn = redis.Redis(
                host=os.getenv('REDIS_HOST', 'localhost'),
                port=int(os.getenv('REDIS_PORT', 6379)),
                db=int(os.getenv('REDIS_DB', 0)),
                decode_responses=True
            )
            
            logging.info("数据库连接初始化成功")
        except Exception as e:
            logging.error(f"数据库连接初始化失败: {str(e)}")
            raise
    
    def check_connections(self):
        """检查数据库连接状态"""
        try:
            if not self.mysql_conn or not self.mysql_conn.is_connected():
                self._init_connections()
            if not self.redis_conn:
                self._init_connections()
            return True
        except Exception as e:
            logging.error(f"检查连接状态时出错: {str(e)}")
            return False
    
    def __del__(self):
        """清理数据库连接"""
        if self.mysql_conn:
            self.mysql_conn.close()
        if self.redis_conn:
            self.redis_conn.close()
    
    def update_item_cf_model(self):
        """更新协同过滤模型"""
        print(f"[{datetime.now()}] 开始更新协同过滤模型...")
        success = self.item_cf.train_model()
        if success:
            print(f"[{datetime.now()}] 协同过滤模型更新成功")
        else:
            print(f"[{datetime.now()}] 协同过滤模型更新失败")
        return success
    
    def update_content_model(self):
        """更新基于内容的推荐模型"""
        print(f"[{datetime.now()}] 开始更新基于内容的推荐模型...")
        success = self.content_recommender.train_model()
        if success:
            print(f"[{datetime.now()}] 基于内容的推荐模型更新成功")
        else:
            print(f"[{datetime.now()}] 基于内容的推荐模型更新失败")
        return success
    
    def update_course_popularity(self):
        """更新课程热度"""
        try:
            logging.info("开始更新课程热度...")
            if not self.check_connections():
                return False
                
            # 获取最近7天的用户行为数据
            query = """
                SELECT course_id, COUNT(*) as view_count
                FROM user_behaviors
                WHERE event_type = 'view'
                AND create_time >= DATE_SUB(NOW(), INTERVAL 7 DAY)
                GROUP BY course_id
            """
            
            cursor = self.mysql_conn.cursor(dictionary=True)
            cursor.execute(query)
            results = cursor.fetchall()
            cursor.close()
            
            if not results:
                logging.info("没有找到课程热度数据")
                return True
                
            # 计算热度分数并保存到Redis
            for result in results:
                course_id = result['course_id']
                view_count = result['view_count']
                
                # 获取课程的其他指标
                cursor = self.mysql_conn.cursor(dictionary=True)
                cursor.execute("""
                    SELECT enroll_count, student_count
                    FROM courses
                    WHERE id = %s
                """, (course_id,))
                course_data = cursor.fetchone()
                cursor.close()
                
                if not course_data:
                    continue
                    
                # 计算热度分数 (浏览数 * 0.4 + 报名数 * 0.4 + 学生数 * 0.2)
                popularity_score = (
                    view_count * 0.4 +
                    course_data['enroll_count'] * 0.4 +
                    course_data['student_count'] * 0.2
                )
                
                # 保存到Redis
                self.redis_conn.zadd('course_popularity', {str(course_id): popularity_score})
            
            logging.info(f"课程热度更新成功，共 {len(results)} 个课程")
            return True
            
        except Exception as e:
            logging.error(f"更新课程热度时出错: {str(e)}")
            return False
    
    def update_frequent_course_combinations(self):
        """更新频繁课程组合"""
        try:
            logging.info("开始更新频繁课程组合...")
            if not self.check_connections():
                return False
                
            # 获取用户购买历史
            query = """
                SELECT user_id, course_id
                FROM user_behaviors
                WHERE event_type = 'purchase'
                AND create_time >= DATE_SUB(NOW(), INTERVAL 30 DAY)
            """
            
            cursor = self.mysql_conn.cursor(dictionary=True)
            cursor.execute(query)
            results = cursor.fetchall()
            cursor.close()
            
            if not results:
                logging.info("没有找到购买历史数据")
                return True
                
            # 构建用户-课程矩阵
            user_courses = {}
            for result in results:
                user_id = result['user_id']
                course_id = result['course_id']
                if user_id not in user_courses:
                    user_courses[user_id] = set()
                user_courses[user_id].add(course_id)
            
            # 转换为DataFrame格式
            all_courses = set()
            for courses in user_courses.values():
                all_courses.update(courses)
            all_courses = sorted(list(all_courses))
            
            data = []
            for user_id, courses in user_courses.items():
                row = {course_id: 1 if course_id in courses else 0 for course_id in all_courses}
                data.append(row)
            
            df = pd.DataFrame(data)
            
            # 使用mlxtend进行频繁项集挖掘
            from mlxtend.frequent_patterns import apriori, association_rules
            
            # 设置最小支持度
            min_support = 0.1
            
            # 生成频繁项集
            frequent_itemsets = apriori(df, min_support=min_support, use_colnames=True)
            
            if len(frequent_itemsets) == 0:
                logging.info("未找到符合条件的频繁项集")
                return True
                
            # 生成关联规则
            rules = association_rules(frequent_itemsets, metric="confidence", min_threshold=0.5)
            
            if len(rules) == 0:
                logging.info("未找到符合条件的关联规则")
                return True
                
            # 保存到Redis
            for _, rule in rules.iterrows():
                antecedents = list(rule['antecedents'])
                consequents = list(rule['consequents'])
                confidence = rule['confidence']
                
                # 为每个前项保存推荐的后项
                for antecedent in antecedents:
                    key = f"course:{antecedent}:recommendations"
                    self.redis_conn.zadd(key, {str(consequent): confidence for consequent in consequents})
            
            logging.info(f"频繁课程组合更新成功，共 {len(rules)} 条规则")
            return True
            
        except Exception as e:
            logging.error(f"更新频繁课程组合时出错: {str(e)}")
            return False
    
    def run_scheduled_updates(self):
        """运行定时更新任务"""
        print(f"[{datetime.now()}] 开始运行定时更新任务")
        
        # 获取模型更新时间
        update_time = RECOMMENDATION_CONFIG.get('model_update_time', '01:00')
        
        # 设置每日更新
        schedule.every().day.at(update_time).do(self._safe_run_update, self.update_all_models)
        
        # 设置每小时更新热度
        schedule.every(1).hours.do(self._safe_run_update, self.update_course_popularity)
        
        # 设置定期检查数据库连接健康状态（每4小时）
        schedule.every(4).hours.do(self._check_db_health)
        
        # 标记为运行中
        self.running = True
        
        # 运行调度器
        while self.running:
            try:
                schedule.run_pending()
                time.sleep(60)  # 每分钟检查一次
            except Exception as e:
                print(f"[{datetime.now()}] 定时任务运行出错: {e}")
                import traceback
                traceback.print_exc()
                
                # 如果发生错误，等待5分钟后继续
                if self.running:
                    time.sleep(300)
    
    def _safe_run_update(self, update_func):
        """安全运行更新任务，捕获所有异常"""
        try:
            return update_func()
        except Exception as e:
            print(f"[{datetime.now()}] 执行更新任务时出错: {e}")
            import traceback
            traceback.print_exc()
            return False
    
    def _check_db_health(self):
        """检查数据库连接健康状态，必要时重新建立连接"""
        print(f"[{datetime.now()}] 检查数据库连接状态...")
        
        try:
            # 检查MySQL连接
            if not self.mysql_conn:
                self.mysql_conn = get_mysql_conn(force_new=True)
                if not self.mysql_conn:
                    print(f"[{datetime.now()}] MySQL连接失败")
                    return False
                
            try:
                with self.mysql_conn.cursor() as cursor:
                    cursor.execute("SELECT 1")
                    cursor.fetchone()
                print(f"[{datetime.now()}] MySQL连接正常")
            except Exception as e:
                print(f"[{datetime.now()}] MySQL连接异常: {e}")
                self.mysql_conn = get_mysql_conn(force_new=True)
                if not self.mysql_conn:
                    print(f"[{datetime.now()}] MySQL重连失败")
                    return False
                print(f"[{datetime.now()}] MySQL连接已重建")
            
            # 检查Redis连接
            if not self.redis_conn:
                self.redis_conn = get_redis_conn(force_new=True)
                if not self.redis_conn:
                    print(f"[{datetime.now()}] Redis连接失败")
                    return False
                
            try:
                self.redis_conn.ping()
                print(f"[{datetime.now()}] Redis连接正常")
            except Exception as e:
                print(f"[{datetime.now()}] Redis连接异常: {e}")
                self.redis_conn = get_redis_conn(force_new=True)
                if not self.redis_conn:
                    print(f"[{datetime.now()}] Redis重连失败")
                    return False
                print(f"[{datetime.now()}] Redis连接已重建")
            
            return True
            
        except Exception as e:
            print(f"[{datetime.now()}] 检查连接状态时出错: {e}")
            return False
    
    def update_all_models(self):
        """更新所有模型"""
        try:
            print("[%s] 开始全量更新所有模型..." % datetime.now())
            
            # 更新协同过滤模型
            cf_success = self.update_item_cf_model()
            
            # 更新基于内容的推荐模型
            content_success = self.update_content_model()
            
            # 更新课程热度
            heat_success = self.update_course_popularity()
            
            # 更新频繁课程组合
            itemset_success = self.update_frequent_course_combinations()
            
            # 更新最后更新时间
            self.update_last_update_time()
            
            print("[%s] 全量更新完成，结果: 协同过滤-%s, 内容推荐-%s, 热度-%s, 频繁组合-%s" % (
                datetime.now(),
                cf_success,
                content_success,
                heat_success,
                itemset_success
            ))
            
            return all([cf_success, content_success, heat_success, itemset_success])
        except Exception as e:
            print(f"更新所有模型时出错: {str(e)}")
            return False
    
    def start(self):
        """启动定时更新任务"""
        if self.scheduler_thread and self.scheduler_thread.is_alive():
            print("定时更新任务已经在运行")
            return False
        
        # 创建并启动调度器线程
        self.scheduler_thread = threading.Thread(target=self.run_scheduled_updates)
        self.scheduler_thread.daemon = True
        self.scheduler_thread.start()
        
        print(f"[{datetime.now()}] 定时更新任务已启动")
        return True
    
    def stop(self):
        """停止定时更新任务"""
        self.running = False
        if self.scheduler_thread:
            self.scheduler_thread.join(timeout=10)
        
        print(f"[{datetime.now()}] 定时更新任务已停止")
        return True

    def load_data(self):
        """从数据库加载数据"""
        try:
            # 获取上次更新时间
            last_update = self.get_last_update_time()
            
            # 构建查询条件
            time_condition = f"AND create_time >= DATE_SUB(NOW(), INTERVAL 90 DAY)"
            if last_update:
                time_condition = f"AND create_time > '{last_update}'"
            
            query = f"""
                SELECT user_id, course_id, event_type, extra_data 
                FROM user_behaviors 
                WHERE event_type IN ('view', 'enroll', 'complete')
                {time_condition}
            """
            
            print("从数据库加载数据...")
            df = pd.read_sql(query, self.mysql_conn)
            
            if not df.empty:
                print(f"样本数据:\n  {df.iloc[0].to_dict()}\n  {df.iloc[1].to_dict()}\n  {df.iloc[2].to_dict()}\n  {df.iloc[3].to_dict()}\n  {df.iloc[4].to_dict()}")
                print(f"从数据库加载了 {len(df)} 条评分数据，涉及 {df['user_id'].nunique()} 个用户")
            else:
                print("没有找到新的用户行为数据")
            
            return df
        except Exception as e:
            print(f"加载数据时出错: {str(e)}")
            return pd.DataFrame()

    def get_last_update_time(self):
        """获取上次更新时间"""
        try:
            # 从Redis获取上次更新时间
            last_update = self.redis_conn.get('last_model_update_time')
            return last_update.decode('utf-8') if last_update else None
        except Exception as e:
            print(f"获取上次更新时间时出错: {str(e)}")
            return None

    def update_last_update_time(self):
        """更新最后更新时间"""
        try:
            self.redis_conn.set('last_model_update_time', datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        except Exception as e:
            print(f"更新最后更新时间时出错: {str(e)}")

    def validate_data(self, df):
        """验证数据有效性"""
        if df.empty:
            print("警告：数据为空")
            return False
        
        # 检查必要字段
        required_columns = ['user_id', 'course_id', 'event_type']
        if not all(col in df.columns for col in required_columns):
            print(f"警告：数据缺少必要字段，当前字段：{df.columns.tolist()}")
            return False
        
        # 检查数据类型
        if not pd.api.types.is_numeric_dtype(df['user_id']) or not pd.api.types.is_numeric_dtype(df['course_id']):
            print("警告：user_id或course_id不是数值类型")
            return False
        
        # 检查事件类型
        valid_events = ['view', 'enroll', 'complete']
        if not df['event_type'].isin(valid_events).all():
            print(f"警告：存在无效的事件类型，有效类型：{valid_events}")
            return False
        
        return True

    def log_update_stats(self, df, model_type):
        """记录更新统计信息"""
        try:
            stats = {
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'model_type': model_type,
                'total_records': len(df),
                'unique_users': df['user_id'].nunique(),
                'unique_courses': df['course_id'].nunique(),
                'event_counts': df['event_type'].value_counts().to_dict()
            }
            
            # 将统计信息保存到Redis
            key = f'model_update_stats:{model_type}:{datetime.now().strftime("%Y%m%d")}'
            self.redis_conn.hmset(key, stats)
            self.redis_conn.expire(key, 7 * 24 * 3600)  # 保存7天
            
            print(f"更新统计信息：{stats}")
        except Exception as e:
            print(f"记录更新统计信息时出错: {str(e)}")

    def update_collaborative_filtering(self):
        """更新协同过滤模型"""
        try:
            print("[%s] 开始更新协同过滤模型..." % datetime.now())
            
            # 加载数据
            df = self.load_data()
            if not self.validate_data(df):
                return False
            
            # 记录更新统计信息
            self.log_update_stats(df, 'collaborative_filtering')
            
            # 更新模型
            # ... 原有的更新逻辑 ...
            
            return True
        except Exception as e:
            print(f"更新协同过滤模型时出错: {str(e)}")
            return False

# 测试代码
if __name__ == "__main__":
    updater = ModelUpdater()
    
    # 执行一次全量更新
    updater.update_all_models()
    
    # 启动定时更新任务
    try:
        updater.start()
        
        # 保持主线程运行
        while True:
            time.sleep(1)
            
    except KeyboardInterrupt:
        # 停止定时更新任务
        updater.stop()
        print("程序已退出") 